package udf.web;

import java.io.IOException;
import java.net.URISyntaxException;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.LinkedList;

import org.apache.pig.EvalFunc;
import org.apache.pig.data.BagFactory;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.DataType;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
import org.apache.pig.impl.logicalLayer.schema.Schema;

import util.webSearch.DeliciousQuery;
import util.webSearch.DeliciousURLInfo;
import util.webSearch.WebResult;

public class fetchDeliciousTags extends EvalFunc<DataBag> {

	BagFactory mBagFactory = BagFactory.getInstance();

	// static ArrayList<String> mTables = new ArrayList<String>();

	// static private final String domain_path =
	// "/user/duartes/helpers/domains.txt";

	public DeliciousURLInfo getDeliciousTags(String url, String path, long delay) throws IOException, URISyntaxException {

		DeliciousURLInfo suggestions = DeliciousQuery
				.parseDeliciousInfoFromURL(url, path);

		try {
			Thread.sleep(delay);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		return suggestions;
	}

	/**
	 * 
	 * Input tuple contains query, max_results, delay
	 * 
	 */
	public DataBag exec(Tuple input) throws IOException {

		DataBag output = mBagFactory.newDefaultBag();

		String url = (String) input.get(0);
		String path = (String) input.get(1);
		String delay_ = (String) input.get(2);

		long delay = Long.valueOf(delay_);

		// System.out.println("Query:" + query);
		// System.out.println("max results:" + max_results);
		DeliciousURLInfo list=null;
		try {
			list = getDeliciousTags(url.trim(), path, delay);
		} catch (URISyntaxException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		Hashtable<String, Long> tags = list.getFrequent_tags();

		if (list == null) {
			TupleFactory mTupleFactory = TupleFactory.getInstance();
			Tuple tuple = mTupleFactory.newTuple();
			tuple.append(url);
			tuple.append("-error-");
			tuple.append(-1);
			// System.out.println(query + "\t" + "-error-" + "\t" +
			// (-1));

		} else if (tags.size() == 0) {
			TupleFactory mTupleFactory = TupleFactory.getInstance();
			Tuple tuple = mTupleFactory.newTuple();
			tuple.append(url);
			tuple.append("-no_results-");
			tuple.append(-2);

		} else {

			Enumeration<String> keys = tags.keys();
			while (keys.hasMoreElements()) {
				String tag = keys.nextElement();
				long freq = tags.get(tag);
				TupleFactory mTupleFactory = TupleFactory.getInstance();
				Tuple tuple = mTupleFactory.newTuple();
				tuple.append(url);
				tuple.append(tag);
				tuple.append((int) freq);
				output.add(tuple);

			}

		}

		return output;

	}

	/**
	 * output schema will have the following format
	 * 
	 * query \t url \t title \t snippet \rank
	 * 
	 * 
	 */
	public Schema outputSchema(Schema input) {
		try {
			Schema bagSchema = new Schema();

			bagSchema.add(new Schema.FieldSchema("url", DataType.CHARARRAY));
			bagSchema.add(new Schema.FieldSchema("tag", DataType.CHARARRAY));
			bagSchema.add(new Schema.FieldSchema("freq", DataType.INTEGER));

			return new Schema(new Schema.FieldSchema(getSchemaName(this
					.getClass().getName().toLowerCase(), input), bagSchema,
					DataType.BAG));

		} catch (Exception e) {
			return null;
		}
	}
}